package com.reger.l2cache.bloomfilter;

import java.util.List;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.util.Assert;

import com.reger.l2cache.bloomfilter.hash.AdditiveHash;
import com.reger.l2cache.bloomfilter.hash.BernsteinHash;
import com.reger.l2cache.bloomfilter.hash.FnvHash;
import com.reger.l2cache.bloomfilter.hash.Md5Hash;
import com.reger.l2cache.bloomfilter.hash.RotatingHash;
import com.reger.l2cache.bloomfilter.hash.RsHash;
import com.reger.l2cache.bloomfilter.hash.SimpleHash;

public class BloomFilter {
	/**
	 * BitSet初始分配2^29个bit ,redis的值最大能存储512MB
	 */
	private static final long DEFAULT_SIZE = 1 << 29;
	private static final String DEFAULT_KEY = "BLOOM_FILTER:DEFAULT";
	/**
	 * 不同哈希函数的种子，一般应取质数，用于处理
	 */
	private static final int[] SEEDS = new int[] { 2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61,
			67, 71, 73, 79, 83, 89, 97 };

	/**
	 * 哈希函数对象
	 */
	private final Hash[] func = new Hash[SEEDS.length];

	private final RedisConnectionFactory connectionFactory;
	private final long size;
	private String key;

	public BloomFilter(RedisConnectionFactory connectionFactory) {
		this(connectionFactory, DEFAULT_KEY, DEFAULT_SIZE);
	}

	public BloomFilter(RedisConnectionFactory connectionFactory, String key) {
		this(connectionFactory, key, DEFAULT_SIZE);
	}

	public BloomFilter(RedisConnectionFactory connectionFactory, String key, long size) {
		Assert.notNull(connectionFactory, "redis链接必须指定");
		Assert.hasLength(key, "redis Key不可以为空");
		Assert.isTrue(size <= DEFAULT_SIZE, "size不能大于 " + DEFAULT_SIZE);
		this.connectionFactory = connectionFactory;
		this.key = key;
		this.size = size;
		this.initHash();
	}

	public void setKey(String key) {
		this.key = key;
	}

	/**
	 * 将字符串标记到bits中
	 * 
	 * @param value string
	 * @return this
	 */
	public BloomFilter add(String value) {
		RedisConnection connection = null;
		try {
			connection = RedisConnectionUtils.getConnection(connectionFactory);
			connection.openPipeline();
			try {
				this.add(connection, value);
			} finally {
				connection.closePipeline();
			}
			return this;
		} finally {
			RedisConnectionUtils.releaseConnection(connection, connectionFactory);
		}
	}

	/**
	 * 将字符串标记到bits中
	 * 
	 * @param values string
	 * @return this
	 */
	public BloomFilter add(String... values) {
		RedisConnection connection = null;
		try {
			connection = RedisConnectionUtils.getConnection(connectionFactory);
			connection.openPipeline();
			try {
				for (int i = 0; i < values.length; i++) {
					String value = values[i];
					this.add(connection, value);
				}
			} finally {
				connection.closePipeline();
			}
			return this;
		} finally {
			RedisConnectionUtils.releaseConnection(connection, connectionFactory);
		}
	}

	private void add(RedisConnection connection, String value) {
		byte[] key = this.key.getBytes();
		for (Hash hash : func) {
			long offset = hash.hash(value);
			connection.setBit(key, offset, true);
		}
	}

	/**
	 * 判断字符串是否已经被bits标记
	 * 
	 * @param value string
	 * @return 是否存在
	 */
	public boolean contains(String value) {
		if (value == null) {
			return false;
		}
		byte[] key = this.key.getBytes();
		RedisConnection connection = null;
		try {
			connection = RedisConnectionUtils.getConnection(connectionFactory);
			connection.openPipeline();
			for (Hash hash : func) {
				connection.getBit(key, hash.hash(value));
			}
			List<Object> list = connection.closePipeline();
			boolean ret = true;
			for (Object object : list) {
				ret &= (boolean) object;
			}
			return ret;
		} finally {
			RedisConnectionUtils.releaseConnection(connection, connectionFactory);
		}
	}

	private void initHash() {
		int i = 0;
		func[i++] = new AdditiveHash(size);
		func[i++] = new BernsteinHash(size);
		func[i++] = new FnvHash(size);
		func[i++] = new Md5Hash(size);
		func[i++] = new RotatingHash(size);
		func[i++] = new RsHash(size);
		for (; i < SEEDS.length; i++) {
			func[i] = new SimpleHash(size, SEEDS[i]);
		}
	}

}